use std::{
    sync::atomic::{AtomicUsize, Ordering},
    sync::mpsc,
    thread,
    time::Duration,
};

use super::*;

// Use an atomic counter to generate unique paths for each test, preventing collisions.
static TEST_ID: AtomicUsize = AtomicUsize::new(0);

fn unique_path() -> String {
    let id = TEST_ID.fetch_add(1, Ordering::SeqCst);
    format!("/shmem_transport_{}_{}", std::process::id(), id)
}

const TEST_BUFFER_SIZE: usize = 1024;

/// Helper function to set up a connected client and server pair.
/// Returns a tuple of (client_transport, server_transport).
fn setup_connection(
    buff_size: usize,
) -> Result<(ShmemEndpoint, ShmemEndpoint), ShmemTransportError> {
    let path = unique_path();

    let server_path = path.clone();
    let server_thread = std::thread::spawn(move || {
        let transport = ShmemTransportBuilder::new(server_path)
            .buff_size(buff_size)
            .connect_timeout(Duration::from_millis(100))
            .build();
        transport.create()
    });

    let client_path = path.clone();
    let client_thread = std::thread::spawn(move || {
        let transport = ShmemTransportBuilder::new(client_path)
            .buff_size(buff_size)
            .connect_timeout(Duration::from_millis(100))
            .build();
        transport.connect()
    });

    Ok((
        server_thread.join().unwrap()?,
        client_thread.join().unwrap()?,
    ))
}

#[test]
fn connect_and_simple_send_receive() {
    const TEST_MSG: &[u8] = b"hello from client";

    let (mut server, mut client) =
        self::setup_connection(TEST_BUFFER_SIZE).expect("Connection error");

    // Client sends a message
    let mut write_guard = client.write(TEST_MSG.len()).unwrap();
    write_guard.copy_from_slice(TEST_MSG);
    write_guard.submit();

    // Server receives the message
    let read_guard = server.read().unwrap();
    assert_eq!(&*read_guard, TEST_MSG);
    read_guard.advance(TEST_MSG.len());
}

#[test]
fn bidirectional_communication() {
    const PING_MSG: &[u8] = b"ping";
    const PONG_MSG: &[u8] = b"pong";

    let (mut server, mut client) =
        self::setup_connection(TEST_BUFFER_SIZE).expect("Connection failed");

    // 1. Client sends "ping"
    let mut write_guard = client.write(PING_MSG.len()).unwrap();
    write_guard.copy_from_slice(PING_MSG);
    write_guard.submit();

    // 2. Server receives "ping"
    let read_guard = server.read().unwrap();
    assert_eq!(&*read_guard, PING_MSG);
    read_guard.advance(PING_MSG.len());

    // 3. Server sends "pong"
    let mut write_guard = server.write(PONG_MSG.len()).unwrap();
    write_guard.copy_from_slice(PONG_MSG);
    write_guard.submit();

    // 4. Client receives "pong"
    let read_guard = client.read().unwrap();
    assert_eq!(&*read_guard, PONG_MSG);
    read_guard.advance(PONG_MSG.len());
}

#[test]
fn multiple_messages_in_sequence() {
    let (mut server, mut client) =
        self::setup_connection(TEST_BUFFER_SIZE).expect("Connection failed");

    for i in 0..1000 {
        let message = format!("message id {}", i);
        let message_bytes = message.as_bytes();

        // Client sends
        let mut write_guard = client.write(message_bytes.len()).unwrap();
        println!("send: {}", message);
        write_guard.copy_from_slice(message_bytes);
        write_guard.submit();

        // Server receives
        let read_guard = server.read().unwrap();
        let read_message = String::from_utf8_lossy(&read_guard);
        println!("recv: {}", read_message);
        assert_eq!(message, read_message, "data corrupted");
        read_guard.advance(message_bytes.len());
    }
}

#[test]
fn ring_buffer_wrap_around() {
    // Use a small buffer to easily force a wrap-around.
    // The actual usable space is smaller due to the control block.
    let buffer_size = 256;
    let (mut server, mut client) = setup_connection(buffer_size).unwrap();

    // Determine usable capacity (total size - control block size - 1 for head/tail logic)
    let usable_capacity = buffer_size - shmem::ControlBlock::size() - 1;

    // 1. Fill the buffer almost to the end to move the head pointer.
    // Let's use 75% of the capacity.
    let msg1_len = (usable_capacity as f64 * 0.75) as usize;
    let msg1 = vec![1u8; msg1_len];

    let mut write_guard = client.write(msg1.len()).unwrap();
    write_guard.copy_from_slice(&msg1);
    write_guard.submit();

    // 2. Server reads the first message, moving the tail pointer forward.
    let read_guard = server.read().unwrap();
    assert_eq!(&*read_guard, msg1.as_slice());
    read_guard.advance(msg1.len());

    // Now, head and tail are both near the end of the buffer.
    // There is not much space left before the physical end of the buffer.

    // 3. Send a second message that is larger than the remaining space at the end,
    // forcing it to wrap around to the beginning of the buffer.
    let msg2_len = (usable_capacity as f64 * 0.5) as usize;
    let msg2 = vec![2u8; msg2_len];

    let mut write_guard = client.write(msg2.len()).unwrap();
    write_guard.copy_from_slice(&msg2);
    write_guard.submit(); // This write should have wrapped around.

    // 4. Server reads the second message. The implementation must correctly
    // reconstruct the fragmented message into a single contiguous slice.
    let read_guard = server.read().unwrap();
    assert_eq!(&*read_guard, msg2.as_slice());
    read_guard.advance(msg2.len());
}

#[test]
fn write_too_large_message_error() {
    let (_server, mut client) =
        self::setup_connection(TEST_BUFFER_SIZE).expect("Connection failed");

    let usable_capacity = TEST_BUFFER_SIZE - shmem::ControlBlock::size() - 1;

    let result = client.write(usable_capacity + 10);
    assert!(matches!(
        result,
        Err(ShmemTransportError::MessageTooLarge { .. })
    ));
}

#[test]
fn connection_closed_error() {
    let (server, mut client) = self::setup_connection(TEST_BUFFER_SIZE).expect("Connection failed");

    // Create a channel to signal when the server has dropped.
    let (tx, rx) = mpsc::channel();
    let client_thread = thread::spawn(move || {
        // Wait for the signal
        rx.recv().unwrap();

        // A short sleep to ensure the OS has processed the state change.
        thread::sleep(Duration::from_millis(10));

        // Now, any operation should fail with ConnectionClosed.
        let read_result = client.read();
        assert!(matches!(
            read_result,
            Err(ShmemTransportError::ConnectionClosed)
        ));
        drop(read_result);

        let write_result = client.write(10);
        assert!(matches!(
            write_result,
            Err(ShmemTransportError::ConnectionClosed)
        ));
        drop(write_result);
    });

    // Drop the server transport. This should trigger the closed state.
    drop(server);
    // Signal the client thread to proceed with its checks.
    tx.send(()).unwrap();

    client_thread.join().unwrap();
}
